package com.yanggu.bigdata.realtime.app.function

import cn.hutool.core.date.DateUtil
import cn.hutool.core.util.StrUtil
import com.alibaba.fastjson.JSONObject
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.ValueStateDescriptor

import java.util.Date

/**
 * kafka中的json字符串
{
    "common":{
        "ar":"310000",
        "ba":"Xiaomi",
        "ch":"xiaomi",
        "is_new":"1",
        "md":"Xiaomi Mix2 ",
        "mid":"mid_5",
        "os":"Android 11.0",
        "uid":"6",
        "vc":"v2.1.134"
    },
    "page":{
        "during_time":4155,
        "last_page_id":"good_detail",
        "page_id":"cart"
    },
    "ts":1608256735000
}
 */
class SetNewUserFlagRichMapFunction extends RichMapFunction[JSONObject, JSONObject] {

  //某个用户首次使用系统的时间字符串
  private lazy val midValueState = getRuntimeContext.getState(
    new ValueStateDescriptor[String]("midValueState", classOf[String]))

  override def map(value: JSONObject): JSONObject = {
    //获取是否新用户的标记
    val isNew = value.getJSONObject("common").getString("is_new")
    //如果是新用户
    if ("1".eq(isNew)) {
      val mid = midValueState.value()
      //如果不为空, 说明已经是老用户了
      if (StrUtil.isBlankIfStr(mid)) {
        //修复错误状态
        value.getJSONObject("common").put("is_new", "0")
      } else {
        midValueState.update(DateUtil.formatDateTime(new Date(value.getLong("ts"))))
      }
    }
    value
  }
}
